package com.sample.api;

import java.lang.reflect.Field;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.TreeMap;

import com.netflix.astyanax.AstyanaxContext;
import com.netflix.astyanax.ColumnListMutation;
import com.netflix.astyanax.Keyspace;
import com.netflix.astyanax.MutationBatch;
import com.netflix.astyanax.connectionpool.NodeDiscoveryType;
import com.netflix.astyanax.connectionpool.OperationResult;
import com.netflix.astyanax.connectionpool.exceptions.ConnectionException;
import com.netflix.astyanax.connectionpool.impl.ConnectionPoolConfigurationImpl;
import com.netflix.astyanax.connectionpool.impl.CountingConnectionPoolMonitor;
import com.netflix.astyanax.impl.AstyanaxConfigurationImpl;
import com.netflix.astyanax.model.Column;
import com.netflix.astyanax.model.ColumnFamily;
import com.netflix.astyanax.model.ColumnList;
import com.netflix.astyanax.model.Row;
import com.netflix.astyanax.model.Rows;
import com.netflix.astyanax.serializers.StringSerializer;
import com.netflix.astyanax.thrift.ThriftFamilyFactory;
import com.sample.mapping.AbstractMapping;
import com.sample.mapping.SingleUpdateMapping;
import com.sample.util.CassandraConfig;
import com.sample.util.ColumnFamilyConfig;

public class CassandraDao extends AbstractDao {

	static Keyspace keyspace;
	static Map<String, ColumnFamily> columnFamilyMap = new HashMap<String, ColumnFamily>();

	static {
		AstyanaxContext<Keyspace> context = new AstyanaxContext.Builder()
				.forCluster(CassandraConfig.CLUSTER_NAME.toString())
				.forKeyspace(CassandraConfig.KEYSPACE_NAME.toString())
				.withAstyanaxConfiguration(
						new AstyanaxConfigurationImpl()
								.setDiscoveryType(NodeDiscoveryType.NONE))
				.withConnectionPoolConfiguration(
						new ConnectionPoolConfigurationImpl(
								CassandraConfig.CONNECTION_POOL_NAME.toString()).setPort(9160)
								.setMaxConnsPerHost(1).setSeeds(CassandraConfig.SEEDS_URL.toString()))
				.withConnectionPoolMonitor(new CountingConnectionPoolMonitor())
				.buildKeyspace(ThriftFamilyFactory.getInstance());

		context.start();
		keyspace = context.getEntity();

		/*
		 * Add column Families to map to be used later while inserting or
		 * getting data from Cassandra.
		 */
		
		//Column family for picture
		ColumnFamily<String, String> picture = new ColumnFamily<String, String>(
				ColumnFamilyConfig.PICTURE.toString(), // Column Family Name
				StringSerializer.get(), // Key Serializer
				StringSerializer.get());
		columnFamilyMap.put(ColumnFamilyConfig.PICTURE.toString(), picture);
		
		//Column family for post
		ColumnFamily<String, String> post = new ColumnFamily<String, String>(
				ColumnFamilyConfig.POST.toString(), // Column Family Name
				StringSerializer.get(), // Key Serializer
				StringSerializer.get());
		columnFamilyMap.put(ColumnFamilyConfig.POST.toString(), post);
		
		//Column family for user line
		ColumnFamily<String, String> user = new ColumnFamily<String, String>(
				ColumnFamilyConfig.USER.toString(), // Column Family Name
				StringSerializer.get(), // Key Serializer
				StringSerializer.get());
		columnFamilyMap.put(ColumnFamilyConfig.USER.toString(), user);
		
		//Column family for user line
		ColumnFamily<String, String> userLine = new ColumnFamily<String, String>(
				ColumnFamilyConfig.USER_LINE.toString(), // Column Family Name
				StringSerializer.get(), // Key Serializer
				StringSerializer.get());
		columnFamilyMap.put(ColumnFamilyConfig.USER_LINE.toString(), userLine);
		
		//Column family for time line
		ColumnFamily<String, String> timeLine = new ColumnFamily<String, String>(
				ColumnFamilyConfig.TIME_LINE.toString(), // Column Family Name
				StringSerializer.get(), // Key Serializer
				StringSerializer.get());
		columnFamilyMap.put(ColumnFamilyConfig.TIME_LINE.toString(), timeLine);
		
		//Column family for follower
		ColumnFamily<String, String> follower = new ColumnFamily<String, String>(
				ColumnFamilyConfig.FOLLOWER.toString(), // Column Family Name
				StringSerializer.get(), // Key Serializer
				StringSerializer.get());
		columnFamilyMap.put(ColumnFamilyConfig.FOLLOWER.toString(), follower);
		
		//Column family for friend
		ColumnFamily<String, String> friend = new ColumnFamily<String, String>(
				ColumnFamilyConfig.FRIEND.toString(), // Column Family Name
				StringSerializer.get(), // Key Serializer
				StringSerializer.get());
		columnFamilyMap.put(ColumnFamilyConfig.FRIEND.toString(), friend);
		
		//Column family for comment
		ColumnFamily<String, String> comment = new ColumnFamily<String, String>(
				ColumnFamilyConfig.COMMENT.toString(), // Column Family Name
				StringSerializer.get(), // Key Serializer
				StringSerializer.get());
		columnFamilyMap.put(ColumnFamilyConfig.COMMENT.toString(), comment);
		
		//Column family for comment mapping
		ColumnFamily<String, String> postCommentMapping = new ColumnFamily<String, String>(
				ColumnFamilyConfig.POST_COMMENT_MAPPING.toString(), // Column Family Name
				StringSerializer.get(), // Key Serializer
				StringSerializer.get());
		columnFamilyMap.put(ColumnFamilyConfig.POST_COMMENT_MAPPING.toString(), postCommentMapping);
		
		//Column family for caption
		ColumnFamily<String, String> caption = new ColumnFamily<String, String>(
				ColumnFamilyConfig.CAPTION.toString(), // Column Family Name
				StringSerializer.get(), // Key Serializer
				StringSerializer.get());
		columnFamilyMap.put(ColumnFamilyConfig.CAPTION.toString(), caption);
		
		//Column family for caption mapping
		ColumnFamily<String, String> postCaptionMapping = new ColumnFamily<String, String>(
				ColumnFamilyConfig.POST_CAPTION_MAPPING.toString(), // Column Family Name
				StringSerializer.get(), // Key Serializer
				StringSerializer.get());
		columnFamilyMap.put(ColumnFamilyConfig.POST_CAPTION_MAPPING.toString(), postCaptionMapping);
	}



	public <T> void get(String columnFamilyName, T id,AbstractMapping<T> mapping) {
		OperationResult<ColumnList<String>> result;
		try {
			result = keyspace.prepareQuery(columnFamilyMap.get(columnFamilyName)).getKey(id).execute();

		Map<String,String> resultMap = new HashMap<String, String>();
		for (Column<String> c : result.getResult()) {
			resultMap.put(c.getName(), c.getStringValue());
		}
		for (Field field : mapping.getClass().getDeclaredFields()) {
			String name = field.getAnnotation(javax.persistence.Column.class).name();
			Object object = null;
			field.setAccessible(true);
				try {
					field.set(mapping,convertToType(resultMap.get(name),field.getType()));
				} catch (IllegalArgumentException e) {
					throw new IllegalStateException("Cassandra connection broken",e);
				} catch (IllegalAccessException e) {
					throw new IllegalStateException("Cassandra connection broken",e);
				}
		}
		mapping.setId(id);
		} catch (ConnectionException e) {
			throw new IllegalStateException("Cassandra connection broken");
		}
		
	}
	
	
	public <T> T convertToType(String src, Class<?> destClass) {
		if(destClass.getName().equals(Boolean.class.getName())) {
			return (T)(Object)Boolean.parseBoolean(src);
		}
		return (T) src;
	}
	
	public <T> void getWithRange(String columnFamilyName, T id,AbstractMapping<T> mapping, String startColumn, String endColumn) {
		getWithRange(columnFamilyName, id, mapping, startColumn, endColumn, 10);
	}
	
	public <T> void getWithRange(String columnFamilyName, T id,AbstractMapping<T> mapping, String startColumn, String endColumn, int noOfRows) {
		OperationResult<ColumnList<String>> result;
		try {
			result = keyspace
					.prepareQuery(columnFamilyMap.get(columnFamilyName))
					.getKey(id).withColumnRange(startColumn, endColumn, false, noOfRows).execute();
		Map<String,String> resultMap = new LinkedHashMap<String, String>();
		for (Column<String> c : result.getResult()) {
			resultMap.put(c.getName(), c.getStringValue());
		}
		for (Field field : mapping.getClass().getDeclaredFields()) {
			field.setAccessible(true);
				try {
					field.set(mapping,resultMap);
				} catch (IllegalArgumentException e) {
					throw new IllegalStateException("Cassandra connection broken");
				} catch (IllegalAccessException e) {
					throw new IllegalStateException("Cassandra connection broken");
				}
		}
		mapping.setId(id);
		} catch (ConnectionException e) {
			throw new IllegalStateException("Cassandra connection broken");
		}
	}
	
	
	public <T> void getAllRow(String columnFamilyName,
			List<AbstractMapping<T>> mappings, int limit,Class<?> mappingClass) {
		try {
			OperationResult<Rows<String, String>> rows = keyspace
					.prepareQuery(columnFamilyMap.get(columnFamilyName))
					.getAllRows()
//			.setRowLimit(limit)
//					// This is the page size
//					.withColumnRange(new RangeBuilder().build())
//					.setExceptionCallback(new ExceptionCallback() {
//						public boolean onException(ConnectionException e) {
//							Assert.fail(e.getMessage());
//						}
//					})
			.execute();
			for (Row<String, String> row : rows.getResult()) {
				 AbstractMapping<T> mapping = (AbstractMapping<T>)mappingClass.newInstance();
				mapping.setId((T) row.getKey());
				Map<String,String> resultMap = new HashMap<String, String>();
				for (Column<String> c : row.getColumns()) {
					resultMap.put(c.getName(), c.getStringValue());
				}
				for (Field field : mapping.getClass().getDeclaredFields()) {
					String name = field.getName();
					Object object = null;
					field.setAccessible(true);
						try {
							field.set(mapping,convertToType(resultMap.get(field.getAnnotation(com.netflix.astyanax.mapping.Column.class).value()),field.getType()));
						} catch (IllegalArgumentException e) {
							throw new IllegalStateException("Cassandra connection broken",e);
						} catch (IllegalAccessException e) {
							throw new IllegalStateException("Cassandra connection broken",e);
						}
				}
				mappings.add(mapping);
			}
		} catch (ConnectionException e) {
			e.printStackTrace();
		} catch (InstantiationException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IllegalAccessException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	//Insert batch
	public <K> void insert(String columnFamilyName,
			AbstractMapping<K> mapping) {
		MutationBatch m = keyspace.prepareMutationBatch();

		ColumnListMutation<String> row = m.withRow(columnFamilyMap.get(columnFamilyName), mapping.getId());

		for (Field field : mapping.getClass().getDeclaredFields()) {
			Object object = null;
			field.setAccessible(true);
			try {
				object = field.get(mapping);
			} catch (IllegalArgumentException e) {
				throw new IllegalStateException(e);
			} catch (IllegalAccessException e) {
				throw new IllegalStateException(e);
			}
			javax.persistence.Column annotations = field
					.getAnnotation(javax.persistence.Column.class);
			if(annotations.name().equals("generateTimeId")){
				LinkedHashMap map = (LinkedHashMap)object;
				Iterator iterator = map.entrySet().iterator();
				while(iterator.hasNext()) {
					Map.Entry<String, String> next = (Entry<String, String>) iterator.next();
					row.putColumn(next.getKey(), next.getValue());
				}
			} else {
				row.putColumn(annotations.name(), object.toString());
			}
		}

		try {
			OperationResult<Void> result = m.execute();
		} catch (ConnectionException e) {
			e.printStackTrace();
			throw new IllegalStateException(e);
		}
	}
	
	//single insert or update
	public <K> void insertOrUpdate(String columnFamilyName, SingleUpdateMapping<K> mapping){
		try {
			
			keyspace.prepareColumnMutation(columnFamilyMap.get(columnFamilyName), mapping.getId(), mapping.getColumnName())
			.putValue(mapping.getData().toString(), null)
			.execute();
		} catch (ConnectionException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}		
	}
	
}
